 1.Kafka高级特性解析之消费者及消费组概念
   
   消费者从订阅的主题消费消息，消费消息的偏移量保存在Kafka的名字是__consumer_offsets 的主题中。
   消费者还可以将自己的偏移量存储到Zookeeper，需要设置offset.storage=zookeeper。
   推荐使用Kafka存储消费者的偏移量。因为Zookeeper不适合高并发。
   多个从同一个主题消费的消费者可以加入到一个消费组中。
   消费组中的消费者共享group_id。
   configs.put("group.id", "xxx");
   group_id一般设置为应用的逻辑名称。比如多个订单处理程序组成一个消费组，可以设置group_id
为"order_process"。
   group_id通过消费者的配置指定： group.id=xxxxx
   消费组均衡地给消费者分配分区，每个分区只由消费组中一个消费者消费。
   一个拥有四个分区的主题，包含一个消费者的消费组。
   此时，消费组中的消费者消费主题中的所有分区。并且没有重复的可能。
   如果在消费组中添加一个消费者2，则每个消费者分别从两个分区接收消息。
   如果消费组有四个消费者，则每个消费者可以分配到一个分区。
   如果向消费组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接收任何消息。
   向消费组添加消费者是横向扩展消费能力的主要方式。
   必要时，需要为主题创建大量分区，在负载增长时可以加入更多的消费者。但是不要让消费者的数量超过主题分
区的数量。
   除了通过增加消费者来横向扩展单个应用的消费能力之外，经常出现多个应用程序从同一个主题消费的情况。
   此时，每个应用都可以获取到所有的消息。只要保证每个应用都有自己的消费组，就可以让它们获取到主题所有
的消息。
   横向扩展消费者和消费组不会对性能造成负面影响。
   为每个需要获取一个或多个主题全部消息的应用创建一个消费组，然后向消费组添加消费者来横向扩展消费能力
和应用的处理能力，则每个消费者只处理一部分消息。
 
 2.心跳机制 
   
   消费者宕机，退出消费组，触发再平衡，重新给消费组中的消费者分配分区。
   由于broker宕机，主题X的分区3宕机，此时分区3没有Leader副本，触发再平衡，消费者4没有对应的主题分区，
则消费者4闲置。
   Kafka 的心跳是 Kafka Consumer 和 Broker 之间的健康检查，只有当 Broker Coordinator 正常时，Consumer
才会发送心跳。
   Consumer 和 Rebalance 相关的2个配置参数：
   参数                      字段
   session.timeout.ms        MemberMetadata.sessionTimeoutMs
   max.poll.interval.ms      MemberMetadata.rebalanceTimeoutMs
   broker 端，sessionTimeoutMs 参数
   broker 处理心跳的逻辑在 GroupCoordinator 类中：如果心跳超期， broker coordinator 会把消费者从 group
中移除，并触发 rebalance。
   private def completeAndScheduleNextHeartbeatExpiration(group: GroupMetadata, member:
MemberMetadata) {
       // complete current heartbeat expectation
       member.latestHeartbeat = time.milliseconds()
       val memberKey = MemberKey(member.groupId, member.memberId)
       heartbeatPurgatory.checkAndComplete(memberKey)
       // reschedule the next heartbeat expiration deadline
       // 计算心跳截止时刻
       val newHeartbeatDeadline = member.latestHeartbeat + member.sessionTimeoutMs
       val delayedHeartbeat = new DelayedHeartbeat(this, group, member,
       newHeartbeatDeadline, member.sessionTimeoutMs)
       heartbeatPurgatory.tryCompleteElseWatch(delayedHeartbeat, Seq(memberKey))
      }
      // 心跳过期
     def onExpireHeartbeat(group: GroupMetadata, member: MemberMetadata,
       heartbeatDeadline: Long) {
          group.inLock {
            if (!shouldKeepMemberAlive(member, heartbeatDeadline)) {
               info(s"Member ${member.memberId} in group ${group.groupId} has failed,
                  removing it from the group")
                 removeMemberAndUpdateGroup(group, member)
             }
       }
  }
   private def shouldKeepMemberAlive(member: MemberMetadata, heartbeatDeadline: Long)
=
    member.awaitingJoinCallback != null ||
      member.awaitingSyncCallback != null ||
      member.latestHeartbeat + member.sessionTimeoutMs > heartbeatDeadline
   consumer 端：sessionTimeoutMs，rebalanceTimeoutMs 参数
   如果客户端发现心跳超期，客户端会标记 coordinator 为不可用，并阻塞心跳线程；如果超过了 poll 消息的间隔
超过了 rebalanceTimeoutMs，则 consumer 告知 broker 主动离开消费组，也会触发 rebalance  
   org.apache.kafka.clients.consumer.internals.AbstractCoordinator.HeartbeatThread
   if (coordinatorUnknown()) {
            if (findCoordinatorFuture != null ||
           lookupCoordinator().failed())
        // the immediate future check ensures that we backoff
        //properly in the case that no
        // brokers are available to connect to.
                 AbstractCoordinator.this.wait(retryBackoffMs);
       } else if (heartbeat.sessionTimeoutExpired(now)) {
        // the session timeout has expired without seeing a
        // successful heartbeat, so we should
        // probably make sure the coordinator is still healthy.
                 markCoordinatorUnknown();
      } else if (heartbeat.pollTimeoutExpired(now)) {
       // the poll timeout has expired, which means that the
       //foreground thread has stalled
       // in between calls to poll(), so we explicitly leave the group.
                 maybeLeaveGroup();
     } else if (!heartbeat.shouldHeartbeat(now)) {
      // poll again after waiting for the retry backoff in case
      //the heartbeat failed or the
      // coordinator disconnected
                AbstractCoordinator.this.wait(retryBackoffMs);
     } else {
                heartbeat.sentHeartbeat(now);
                sendHeartbeatRequest().addListener(new
        RequestFutureListener<Void>() {
      @Override
      public void onSuccess(Void value) {
           synchronized (AbstractCoordinator.this) {
               heartbeat.receiveHeartbeat(time.milliseconds());
      }
    }
      @Override
      public void onFailure(RuntimeException e) {
          synchronized (AbstractCoordinator.this) {
             if (e instanceof
                    RebalanceInProgressException) {
      // it is valid to continue heartbeating
      //while the group is rebalancing. This
      // ensures that the coordinator keeps the
      //member in the group for as long
      // as the duration of the rebalance
      //timeout. If we stop sending heartbeats,
      // however, then the session timeout may
      //expire before we can rejoin.
           heartbeat.receiveHeartbeat(time.milliseconds());
              } else {
                      heartbeat.failHeartbeat();
    // wake up the thread if it's sleeping to reschedule the heartbeat
                      AbstractCoordinator.this.notify();
              }
           }
        }
    });
  }